package com.example.controller;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author zhy
 * @version 1.0.0
 * @ClassName KafkaConsumer.java
 * @Description TODO
 * @createTime 2021年07月03日 15:54:00
 */
@Component
public class KafkaConsumer {

    /**
     * 最普通的消费
     * @param record
     */
    @KafkaListener(topics = {"topic1"})
    public void onMessage1(ConsumerRecord<?, ?> record) {
        System.out.println("简单消费：topic=" + record.topic() + ", partition=" + record.partition() +
                ", value=" + record.value() + ", key=" + record.key() + ", offset=" + record.offset());
        record.headers().forEach(i -> System.out.println(i));
    }

    /**
     * 指定消费异常的处理bean的消费
     * @param record
     */
    @KafkaListener(topics = {"topic2"}, errorHandler = "listenErrorHandler")
    public void onMessage2(ConsumerRecord<?, ?> record) {
        System.out.println("异常消费：topic=" + record.topic() + ", partition=" + record.partition() +
                ", value=" + record.value() + ", key=" + record.key());
        throw new NullPointerException("测试错误处理器");
    }

    /**
     * id：消费者ID
     * groupId：消费组ID
     * topicPartitions：可配置更加详细的监听信息，可指定topic、partition、offset监听
     * 注意topics和topicPartitions不能同时使用
     * 该方法监听的含义：监听topic1的0号分区，同时监听topic2的0号分区和topic2的1号分区里面offset从8开始的消息
     * @param record
     */
    @KafkaListener(id = "consumer1", groupId = "felix-group",
            topicPartitions = {
                @TopicPartition(topic = "topic1", partitions = {"0"}),
                @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
            })
    public void onMessage3(ConsumerRecord<?, ?> record) {
        System.out.println("复杂消费：topic:" + record.topic() + ", partition:" + record.partition() +
                ", offset:" + record.offset() + ", value:" + record.value());
    }

    @KafkaListener(id = "consumer2", groupId = "felix-group", topics = "topic3")
    public void onMessage4(List<ConsumerRecord<?, ?>> records) {
        System.out.println("批量消费一次，records.size()=" + records.size());
        for (ConsumerRecord<?, ?> record : records) {
            System.out.println(record.value());
        }
    }

    @Bean
    public ConsumerAwareListenerErrorHandler listenErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {
            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
                System.out.println("message=" + message.getPayload());
                System.out.println("exception=" + exception.getMessage());
                System.out.println("consumer=" + consumer.toString());
                consumer.listTopics().forEach((k, v) -> {
                    System.out.println("key=" + k + ", value=" + v);
                });
                System.out.println("message.headers=" + JSONObject.toJSONString(message.getHeaders()));
                return null;
            }
        };
    }
}
